package beautiful.butterfly.drds.data_exchange.transport.exchanger;

import beautiful.butterfly.drds.data_exchange.exception.CommonErrorCode;
import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.plugin.TaskPluginCollector;
import beautiful.butterfly.drds.data_exchange.statistics.communication.Communication;
import beautiful.butterfly.drds.data_exchange.transport.record.DefaultRecord;
import beautiful.butterfly.drds.data_exchange.transport.record.TerminateRecord;
import beautiful.butterfly.drds.data_exchange.transport.record.column.Record;
import beautiful.butterfly.drds.data_exchange.transport.record_queue.RecordQueue;
import beautiful.butterfly.drds.data_exchange.transport.transformer.TransformerExecution;
import beautiful.butterfly.drds.data_exchange.util.Configuration;

import java.util.List;

public class Exchanger extends TransformerExchanger implements Produce, Consume

{


    private RecordQueue recordQueue;
    private Configuration configuration;
    private volatile boolean shutdown = false;

    @SuppressWarnings("unchecked")
    public Exchanger(final int taskGroupId, final int taskId, final RecordQueue recordQueue, final Communication communication, List<TransformerExecution> transformerExecs, final TaskPluginCollector pluginCollector)
    {
        super(taskGroupId, taskId, communication, transformerExecs, pluginCollector);
        assert recordQueue != null;
        this.recordQueue = recordQueue;
        this.configuration = recordQueue.getConfiguration();

    }


    public Record getFromRecordList()
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }
        Record record = this.recordQueue.pull();
        return (record instanceof TerminateRecord ? null : record);
    }


    public Record createRecord()
    {
        return new DefaultRecord();
    }


    public void addToRecordList(Record record)
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }
        record = doTransformer(record);
        if (record == null)
        {
            return;
        }
        this.recordQueue.push(record);
        //和channel的统计保持同步
        doStat();
    }


    public void pushAllToRecordQueue()
    {
    }


    public void terminate()
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }
        this.recordQueue.pushTerminateRecord(TerminateRecord.get());
        //和channel的统计保持同步
        doStat();
    }


    public void shutdown()
    {
        shutdown = true;
    }
}
